package com.shenhan.datasource.crawler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.shenhan.datasource.bean.CovidBean;
import com.shenhan.datasource.utils.HttpUtils;
import com.shenhan.datasource.utils.TimeUtil;
import org.apache.commons.lang3.StringUtils;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @Author shenhan
 * @Date: 2020/9/7 22:51
 * @Description: 实现疫情爬虫
 */
@Component
public class Covid19DataCrawler {
    @Autowired
    private KafkaTemplate kafkaTemplate;

    @Scheduled(cron = "0 0 8 * * ?")
    public void crawler() {
        String dateTime = TimeUtil.format(System.currentTimeMillis(), "yyyy-MM-dd");
        String html = HttpUtils.getHtml("http://ncov.dxy.cn/ncovh5/view/pneumonia");
        Document document = Jsoup.parse(html);
        String text = document.select("script[id=getAreaStat]").toString();
        //使用正则获取疫情json数据
        String pattern = "\\[(.*)]";
        Pattern compile = Pattern.compile(pattern);
        Matcher matcher = compile.matcher(text);
        String json;
        if (matcher.find()) {
            json = matcher.group(0);
        } else {
            json = "no match";
        }
        List<CovidBean> covidBeans = JSON.parseArray(json, CovidBean.class);
        covidBeans.forEach(bean -> {
            bean.setDateTime(dateTime);
            String cities = bean.getCities();
            if (StringUtils.isNotBlank(cities)) {
                List<CovidBean> citiesList = JSON.parseArray(cities, CovidBean.class);
                citiesList.forEach(c -> {
                    c.setDateTime(dateTime);
                    c.setPid(bean.getLocationId());
                    c.setProvinceShortName(bean.getProvinceShortName());
                    String data = JSON.toJSONString(c);
                    kafkaTemplate.send("covid19", c.getPid(), data);
                });
            }

            if (StringUtils.isNotBlank(bean.getStatisticsData())) {
                String staticsData = HttpUtils.getHtml(bean.getStatisticsData());
                JSONObject jsonObject = JSON.parseObject(staticsData);
                String staticStr = jsonObject.getString("data");
                bean.setStatisticsData(staticStr);
                bean.setCities(null);
                String data = JSON.toJSONString(bean);
                kafkaTemplate.send("covid19", bean.getLocationId(), data);
            }
        });
    }
}
